/*
 *  Copyright 2013 Utkin Dmitry
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <QDebug>
#include <QSqlQuery>
#include <QSqlError>
#include <QHostAddress>
#include <QSettings>
#include <QHash>
#include <QSet>
#include <protoqtrpc/RpcController.h>
#include <protoqtrpc/DoneClosureRunner.h>
#include <protoqtrpc/SocketRpcServerSession.h>
#include "Settings.h"
#include "LocationDb.h"
#include "LocationServiceImpl.h"

namespace protoqtrpc
{
namespace location
{

class PortsAllocator
{
public:
    static PortsAllocator& inst()
    {
        static PortsAllocator instance;
        return instance;
    }

    quint16 allocate(quint32 host)
    {
        quint16 port = mMinPort;

        QSet<quint16>& ports = mHosts[host];
        for (port = mMinPort; port <= mMaxPort && ports.contains(port); ++port);

        if (port == mMaxPort)
        {
            qWarning() << "No more free ports for host: " << QHostAddress(host).toString();
            return 0;
        }

        ports.insert(port);

        return port;
    }

    void deallocate(quint32 host, quint16 port)
    {
        QHash<quint32, QSet<quint16> >::Iterator itHost = mHosts.find(host);
        if (itHost != mHosts.end())
        {
            QSet<quint16>& ports = *itHost;
            ports.remove(port);
        }
    }

private:
    PortsAllocator()
    {
        QSettings& settings = Settings::settings();

        bool ok = false;
        mMinPort = static_cast<quint16>(settings.value("ports/min", 33701).toUInt(&ok));
        if (!ok)
            qDebug() << "using default ports/min setting: " << mMinPort;

        mMaxPort = static_cast<quint16>(settings.value("ports/max", 33800).toUInt(&ok));
        if (!ok)
            qDebug() << "using default ports/max setting: " << mMaxPort;


        // read allocated ports from db

        QSqlQuery query(LocationDb::inst().db());

        query.prepare("SELECT host, port FROM servers ORDER BY host");

        if (!query.exec())
        {
            qWarning() << "Failed to get ports";
            return;
        }

        quint32 host = 0;
        quint16 port = 0;
        while (query.next())
        {
            host = static_cast<quint32>(query.value(0).toUInt());
            if (!host)
            {
                qWarning() << "invalid host readed";
                continue;
            }

            port = static_cast<quint16>(query.value(1).toUInt());
            if (port < mMinPort || port > mMaxPort)
                qWarning() << "cached port (" << port
                           << ") is out of range of allowed: " << mMinPort << "-" << mMaxPort;

            mHosts[host].insert(port);
        }
    }

private:
    quint16 mMinPort;
    quint16 mMaxPort;
    QHash<quint32, QSet<quint16> > mHosts;
};

LocationServiceImpl::LocationServiceImpl()
{
}

LocationServiceImpl::~LocationServiceImpl()
{
}

void LocationServiceImpl::registerServer(::google::protobuf::RpcController* controller,
                                          const ServerRegistrationRequest* request,
                                          ServerRegistrationResponse* response,
                                          ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;

    quint16 port = 0;

    QSqlQuery query(LocationDb::inst().db());

    // query for port
    query.prepare("SELECT port FROM servers WHERE name = :server AND host = :host");
    query.bindValue(":server", QString::fromUtf8(request->name().c_str()));
    query.bindValue(":host", host);

    if (!query.exec())
        qWarning() << "Query execution failed: " << query.lastError();
    else if (query.next())
    {
        response->set_port(static_cast<quint16>(query.value(0).toUInt()));
        response->set_status(STATUS_ERROR_ALREADY_REGISTERED);
        return;
    }

    query.clear();


    port = PortsAllocator::inst().allocate(host);
    if (!port)
    {
        qWarning() << "No more free ports for host: " << QHostAddress(host).toString();
        response->set_status(STATUS_ERROR_NO_FREE_PORTS);
        return;
    }

    query.prepare("INSERT INTO servers(name, description, host, port) "
                  "VALUES (:name, :desc, :host, :port)");

    query.bindValue(":name", QString::fromUtf8(request->name().c_str()));
    query.bindValue(":desc", QString::fromUtf8(request->description().c_str()));
    query.bindValue(":host", host);
    query.bindValue(":port", port);

    if (!query.exec())
    {
        const QSqlError& error = query.lastError();
        qWarning() << "query exec fail: " << error.text();
        response->set_status(STATUS_ERROR_OTHER);

        PortsAllocator::inst().deallocate(host, port);
        return;
    }

    response->set_port(port);
    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::unregisterServer(::google::protobuf::RpcController* controller,
                                            const NameRequest* request,
                                            GenericResponse* response,
                                            ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(QString::fromUtf8(request->name().c_str()));
    quint16 port = 0;


    // query for port
    QSqlQuery query(LocationDb::inst().db());

    query.prepare("SELECT port FROM servers WHERE name = :server AND host = :host");
    query.bindValue(":server", server);
    query.bindValue(":host", host);

    if (!query.exec())
        qWarning() << "Query execution failed: " << query.lastError();
    else
        if (query.next())
            port = static_cast<quint16>(query.value(0).toUInt());

    query.clear();


    query.prepare("DELETE FROM servers WHERE name = :server AND host = :host");
    query.bindValue(":server", server);
    query.bindValue(":host", host);
    query.exec();

    if (!port)
    {
        response->set_status(STATUS_ERROR_NOT_REGISTERED);
        return;
    }

    PortsAllocator::inst().deallocate(host, port);
    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::getServerInfo(::google::protobuf::RpcController* controller,
                                         const NameRequest* request,
                                         ServerInfoResponse* response,
                                         ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(request->name().c_str());


    // query for port
    QSqlQuery query(LocationDb::inst().db());

    query.prepare("SELECT name, description, host, port, time FROM servers "
                  "WHERE name = :server AND host = :host");
    query.bindValue(":server", server);
    query.bindValue(":host", host);

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    if (!query.next())
    {
        qWarning() << "Server " << server << " is not registered on host "
                   << QHostAddress(host).toString();
        response->set_status(STATUS_ERROR_NOT_REGISTERED);
        return;
    }

    ServerInfo& serverInfo = *response->mutable_serverinfo();

    const QByteArray& name = query.value(0).toString().toUtf8();
    serverInfo.set_name(name.constData(), name.size());

    const QByteArray& description = query.value(1).toString().toUtf8();
    serverInfo.set_description(description.constData(), description.size());

    serverInfo.set_host(query.value(2).toUInt());
    serverInfo.set_port(query.value(3).toUInt());
    serverInfo.set_time(query.value(4).toUInt());

    response->set_status(STATUS_SUCCESS);
}

void LocationServiceImpl::getServerServicesInfo(::google::protobuf::RpcController* controller,
                                                 const NameRequest* request,
                                                 ServerServicesInfoResponse* response,
                                                 ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(QString::fromUtf8(request->name().c_str()));


    // query for port
    QSqlQuery query(LocationDb::inst().db());

    query.prepare("SELECT name, name, description, time FROM servers "
                  "WHERE name = :server AND host = :host");
    query.bindValue(":server", server);
    query.bindValue(":host", host);

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    if (!query.next())
    {
        qWarning() << "Server " << server << " is not registered on host "
                   << QHostAddress(host).toString();
        response->set_status(STATUS_ERROR_NOT_REGISTERED);
        return;
    }

    ServerServicesInfo& serverServicesInfo = *response->mutable_serverinfo();

    // fill in the server info
    int serverId = query.value(0).toInt();

    {
        ServerInfo& serverInfo = *serverServicesInfo.mutable_serverinfo();

        const QByteArray& name = query.value(1).toString().toUtf8();
        serverInfo.set_name(name.constData(), name.size());

        const QByteArray& description = query.value(2).toString().toUtf8();
        serverInfo.set_description(description.constData(), description.size());

        serverInfo.set_time(query.value(3).toUInt());
    }

    query.clear();


    // fill in the services info
    ::google::protobuf::RepeatedPtrField< ServiceInfo >& services =
            *serverServicesInfo.mutable_servicesinfo();

    query.prepare("SELECT name, description, time FROM servers "
                  "WHERE server_id = :server_id");
    query.bindValue(":server_id", serverId);

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    while (query.next())
    {
        ServiceInfo& serviceInfo = *services.Add();

        const QByteArray& name = query.value(0).toString().toUtf8();
        serviceInfo.set_name(name.constData(), name.size());

        const QByteArray& description = query.value(1).toString().toUtf8();
        serviceInfo.set_description(description.constData(), description.size());

        serviceInfo.set_time(query.value(2).toUInt());
    }

    response->set_status(STATUS_SUCCESS);
}

void LocationServiceImpl::getServersInfo(::google::protobuf::RpcController* /*controller*/,
                                          const VoidRequest* /*request*/,
                                          ServersInfoResponse* response,
                                          ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    QSqlQuery queryServer(LocationDb::inst().db());
    QSqlQuery queryService(LocationDb::inst().db());

    queryServer.prepare("SELECT name, name, description, host, port, time FROM servers");
    queryService.prepare("SELECT name, description, time FROM services "
                         "WHERE server_id = :server_id");

    if (!queryServer.exec())
    {
        qWarning() << "Query execution failed: " << queryServer.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    ::google::protobuf::RepeatedPtrField< ServerServicesInfo >& servers =
            *response->mutable_servers();

    while (queryServer.next())
    {
        // fill in the server info
        ServerServicesInfo& serverServicesInfo = *servers.Add();

        int serverId = queryServer.value(0).toInt();
        {
            ServerInfo& serverInfo = *serverServicesInfo.mutable_serverinfo();

            const QByteArray& name = queryServer.value(1).toString().toUtf8();
            serverInfo.set_name(name.constData(), name.size());

            const QByteArray& description = queryServer.value(2).toString().toUtf8();
            serverInfo.set_description(description.constData(), description.size());

            serverInfo.set_host(static_cast<quint32>(queryServer.value(3).toUInt()));
            serverInfo.set_port(static_cast<quint16>(queryServer.value(4).toUInt()));
            serverInfo.set_time(queryServer.value(5).toUInt());
        }


        // now, query for server's services
        queryService.bindValue(":server_id", serverId);
        if (!queryService.exec())
        {
            qWarning() << "Query execution failed: " << queryService.lastError().text();
            response->set_status(STATUS_ERROR_OTHER);
            return;
        }

        ::google::protobuf::RepeatedPtrField< ServiceInfo >& services =
                *serverServicesInfo.mutable_servicesinfo();

        while (queryService.next())
        {
            ServiceInfo& serviceInfo = *services.Add();

            const QByteArray& name = queryService.value(0).toString().toUtf8();
            serviceInfo.set_name(name.constData(), name.size());

            const QByteArray& description = queryService.value(1).toString().toUtf8();
            serviceInfo.set_description(description.constData(), description.size());

            serviceInfo.set_time(queryService.value(2).toUInt());
        }
    }

    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::registerService(::google::protobuf::RpcController* controller,
                                          const ServiceRegistrationRequest* request,
                                          GenericResponse* response,
                                          ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(QString::fromUtf8(request->servername().c_str()));

    QSqlQuery query(LocationDb::inst().db());

    int serverId = -1;
    Status status = getServerId(query, host, server, serverId);
    if (status != STATUS_SUCCESS)
    {
        response->set_status(status);
        return;
    }

    query.prepare("INSERT INTO services(server_id, name, description) "
                  "VALUES (:server_id, :name, :description)");
    query.bindValue(":server_id", serverId);
    query.bindValue(":name", QString::fromUtf8(request->servicename().c_str()));
    query.bindValue(":description", QString::fromUtf8(request->description().c_str()));

    if (!query.exec())
    {
        if (query.lastError().number() == 19) // constraint failed
        {
            response->set_status(STATUS_ERROR_ALREADY_REGISTERED);
        }
        else
        {
            qWarning() << "Query execution failed: " << query.lastError();
            response->set_status(STATUS_ERROR_OTHER);
        }
        return;
    }

    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::unregisterService(::google::protobuf::RpcController* controller,
                                            const ServiceId* request,
                                            GenericResponse* response,
                                            ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(QString::fromUtf8(request->servername().c_str()));

    QSqlQuery query(LocationDb::inst().db());

    int serverId = -1;
    Status status = getServerId(query, host, server, serverId);
    if (status != STATUS_SUCCESS)
    {
        response->set_status(status);
        return;
    }

    query.prepare("DELETE FROM services WHERE server_id = :server_id AND name = :name");
    query.bindValue(":server_id", serverId);
    query.bindValue(":name", QString::fromUtf8(request->servername().c_str()));

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::getServiceInfo(::google::protobuf::RpcController* controller,
                                         const ServiceId* request,
                                         ServiceInfoResponse* response,
                                         ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    RpcController* rpcController = static_cast<RpcController*>(controller);
    const SocketPeerInfo* peerInfo =
            static_cast<const SocketPeerInfo*>(rpcController->peerInfo());
    if (peerInfo == NULL)
    {
        qWarning() << "Failed to get peer info";
        return;
    }

    quint32 host = peerInfo->address;
    QString server(QString::fromUtf8(request->servername().c_str()));

    QSqlQuery query(LocationDb::inst().db());

    int serverId = -1;
    Status status = getServerId(query, host, server, serverId);
    if (status != STATUS_SUCCESS)
    {
        response->set_status(status);
        return;
    }

    query.prepare("SELECT name, description, time FROM services "
                  "WHERE server_id = :server_id AND name = :name");
    query.bindValue(":server_id", serverId);
    query.bindValue(":name", QString::fromUtf8(request->servicename().c_str()));

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    if (!query.next())
    {
        qWarning() << "Service " << request->servicename().c_str()
                   << " is not registered on server " << server
                   << " on host " << QHostAddress(host).toString();
        response->set_status(STATUS_ERROR_NOT_REGISTERED);
        return;
    }

    ServiceInfo& serviceInfo = *response->mutable_info();

    const QByteArray& name = query.value(0).toString().toUtf8();
    serviceInfo.set_name(name.constData(), name.size());

    const QByteArray& description = query.value(1).toString().toUtf8();
    serviceInfo.set_description(description.constData(), description.size());

    serviceInfo.set_time(query.value(2).toUInt());

    response->set_status(STATUS_SUCCESS);
}


void LocationServiceImpl::locateService(::google::protobuf::RpcController* /*controller*/,
                                        const LocateServiceRequest* request,
                                        LocateServiceResponse* response,
                                        ::google::protobuf::Closure* done)
{
    QMutexLocker l(&mMutex);
    DoneClosureRunner r(done);

    QSqlQuery query(LocationDb::inst().db());
    QString queryStr = "SELECT s.name, s.description, s.time, "
            "c.name, c.description, c.host, c.port, c.time "
            "FROM services s INNER JOIN servers c ON s.server_id = c.id "
            "WHERE s.name = :name";

    if (request->has_serverhost())
        queryStr += " AND c.host = :server_host";

    if (request->has_servername())
        queryStr += " AND c.name = :server_name";

    query.prepare(queryStr);
    query.bindValue(":name", QString::fromUtf8(request->servicename().c_str()));

    if (request->has_serverhost())
        query.bindValue(":server_host", request->serverhost());

    if (request->has_servername())
        query.bindValue(":server_name", QString::fromUtf8(request->servername().c_str()));


    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        response->set_status(STATUS_ERROR_OTHER);
        return;
    }

    ::google::protobuf::RepeatedPtrField< ServiceLocation >& locations =
            *response->mutable_locations();
    while (query.next())
    {
        ServiceLocation& location = *locations.Add();
        ServiceInfo& serviceInfo = *location.mutable_serviceinfo();

        const QByteArray& serviceName = query.value(0).toString().toUtf8();
        serviceInfo.set_name(serviceName.constData(), serviceName.size());
        const QByteArray& serviceDesc = query.value(1).toString().toUtf8();
        serviceInfo.set_description(serviceDesc.constData(), serviceDesc.size());
        serviceInfo.set_time(static_cast<quint32>(query.value(2).toUInt()));

        ServerInfo& serverInfo = *location.mutable_serverinfo();
        const QByteArray& serverName = query.value(3).toString().toUtf8();
        serverInfo.set_name(serverName.constData(), serverName.size());
        const QByteArray& serverDesc = query.value(4).toString().toUtf8();
        serverInfo.set_description(serverDesc.constData(), serverDesc.size());
        serverInfo.set_host(static_cast<quint32>(query.value(5).toUInt()));
        serverInfo.set_port(static_cast<quint16>(query.value(6).toUInt()));
        serverInfo.set_time(static_cast<quint32>(query.value(7).toUInt()));
    }

    response->set_status(STATUS_SUCCESS);
}

Status LocationServiceImpl::getServerId(QSqlQuery& query, quint32 host, const QString& server,
                                         int& serverId)
{
    query.prepare("SELECT id FROM servers WHERE name = :server AND host = :host");
    query.bindValue(":server", server);
    query.bindValue(":host", host);

    if (!query.exec())
    {
        qWarning() << "Query execution failed: " << query.lastError();
        return STATUS_ERROR_OTHER;
    }

    if (!query.next())
    {
        qWarning() << "Server " << server << " is not registered on host "
                   << QHostAddress(host).toString();
        return STATUS_ERROR_NOT_REGISTERED;
    }

    serverId = query.value(0).toInt();

    query.clear();
    return STATUS_SUCCESS;
}


}
}
